package com.markhsiu.MiniMQ.client.consumer;

import com.markhsiu.MiniMQ.client.CallBackor;
import com.markhsiu.MiniMQ.client.Client;
import com.markhsiu.MiniMQ.model.Message;
import com.markhsiu.MiniMQ.model.Result;
import com.markhsiu.MiniMQ.model.constant.MessageCmdEnum;
import com.markhsiu.MiniMQ.model.constant.MessageSourceEnum;
import com.markhsiu.MiniMQ.remote.transport.netty.NettyClient;

/**
 * Created by Mark Hsiu on 2017/2/8.
 */
public class Consumer {

	private Client client = new NettyClient();
	private String consumerID;
	  
	@SuppressWarnings("unused")
	private Consumer() {}
	
	
	public Consumer(CallBackor callBackor) {
		client.setCallBackor(callBackor);
	}
	
	public Consumer(CallBackor callBackor,Client client) {
		this(callBackor);
		this.client = client;
		
	}

	public void start(final String consumerID) {
		this.consumerID = consumerID;
		client.open();
		client.connect();
		
	}

	public boolean register(){
		Message message = new Message();
		message.setMessageID(consumerID);
		message.setConsumerID(consumerID);
		message.setSource(MessageSourceEnum.CUSTOMER);
		message.setTarget(MessageSourceEnum.BROKER);
		message.setCmd(MessageCmdEnum.CONNECT);
		message.setTimestamp(System.currentTimeMillis());
		Result result = client.send(message);
		if(result.getError() == 0 && result.getMessage().getCmd() == MessageCmdEnum.SUCCESS){
			return true;
		}
		
		return false;
	}
	
	public boolean subscribe(String topic){
		Message message = new Message();
		message.setMessageID(consumerID);
		message.setConsumerID(consumerID);
		message.setTopic(topic);
		message.setSource(MessageSourceEnum.CUSTOMER);
		message.setTarget(MessageSourceEnum.BROKER);
		message.setCmd(MessageCmdEnum.SUB);
		message.setTimestamp(System.currentTimeMillis());
		Result result = client.send(message);
		if(result.getError() == 0 && result.getMessage().getCmd() == MessageCmdEnum.SUCCESS){
			return true;
		}
		
		return false;
	}
	
	public boolean unsubscribe(String topic){
		Message message = new Message();
		message.setMessageID(consumerID);
		message.setConsumerID(consumerID);
		message.setTopic(topic);
		message.setSource(MessageSourceEnum.CUSTOMER);
		message.setTarget(MessageSourceEnum.BROKER);
		message.setCmd(MessageCmdEnum.UNSUB);
		message.setTimestamp(System.currentTimeMillis());
		Result result = client.send(message);
		if(result.getError() == 0 && result.getMessage().getCmd() == MessageCmdEnum.SUCCESS){
			return true;
		}
		
		return false;
	}
 

	public void end() {
		client.close();
	}
	
}
